package com.my.study.java.serialization;

import com.my.study.java.bean.Company;
import org.apache.kafka.common.serialization.Serializer;

import java.io.UnsupportedEncodingException;
import java.nio.ByteBuffer;
import java.util.Map;

/**
 * {@link Company}序列化
 *
 * @author: yidujun
 * @create: 2021/07/23 14:03
 */
public class CompanySerializer implements Serializer<Company> {

    private String defaultEncode = "UTF-8";

    @Override
    public void configure(Map<String, ?> configs, boolean isKey) {

    }

    @Override
    public byte[] serialize(String topic, Company data) {
        if (null == data) {
            return null;
        }
        byte[] name, address;
        try {
            if (data.getName() != null) {
                name = data.getName().getBytes(defaultEncode);
            } else {
                name = new byte[0];
            }

            if (null != data.getAddress()) {
                address = data.getAddress().getBytes(defaultEncode);
            } else {
                address = new byte[0];
            }

            ByteBuffer buffer = ByteBuffer.allocate(4 + 4 + name.length + address.length);
            buffer.putInt(name.length);
            buffer.put(name);
            buffer.putInt(address.length);
            buffer.put(address);

            return buffer.array();
        } catch (Exception e) {
            e.printStackTrace();
        }
        return new byte[0];
    }

    @Override
    public void close() {

    }


    public static void main(String[] args) throws UnsupportedEncodingException {
        CompanySerializer companySerializer = new CompanySerializer();
        Company company = new Company();
        company.setName("ths");
        company.setAddress("同顺街1号");

        byte[] bytes = companySerializer.serialize("", company);

        String s = new String(bytes, "UTF-8");
        System.out.println(s);
    }
}
